package com.dxf.bigdata.D04_spark

import org.apache.commons.logging.{Log, LogFactory}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *  RDD方法主要有 转换 和 行动
 *   转换 如  Map , flatMap
 *   行动 collect
 *
 *
 *   RDD算子:   RDD的方法就是算子 --> 对数据的操作称为算子
 *
 *   RDD算子 => 转换算子 Map , 行动算子  collect
 *
 *   ===========================
 *
 *
 *   Map  => 处理数据,一个处理完了处理下一个 ==> 效率肯定底 (对比字节流) ==> 处理方式 缓存区
 *
 *
 *
 *  =====================================
 *
 *
 *   glom : 将同一个分区的数据直接转换为相同类型的内存数组进行处理, 分区不变
 *
 *    [1 ,2 ] [3 ,4 ] --> [List(1,2)] [List(3,4)]
 *
 *
 *   合并 RDD[Int]  -> RDD[Array[Int]]
 *
 *
 *   ===========================================
 *
 *
 *    groupBy 分组 ,同一组放在一起
 *    groupBy 会将数据打乱重新组合,我们将这样的操作称之为shuffle,极限情况下,数据可能被分再同一个分区中
 *
 *
 */
object T14_groupBy {
  private val log: Log = LogFactory.getLog(T08_RDD使用_从日志中读取uri.getClass)

  def main(args: Array[String]): Unit = {

    //TODO 准备环境
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val sc = new SparkContext(sparkConf)


    //TODO 创建RDD
    val rdd: RDD[Int] = sc.makeRDD(List(1, 2, 3, 4),2)
    //groupBy 将数据源中的每一个数据进行分组判断,根据返回的分组key进行分组,相同key值的数据会放在一个组中

    //按照返回值分组
    //如 List(1, 2, 3, 4) , 返回值 1,2,3,4 则4组 1,2,3,4组
    def groupFunction(num:Int)={
      num
    }
    // List(1, 2, 3, 4) 返回值  1, 0,1,0 则2组 0组 和 1组
    def groupFunction2(num:Int)={
      num%2
    }

    /**
     * (0,CompactBuffer(2, 4))
     * (1,CompactBuffer(1, 3))
     */
    val rddGroup: RDD[(Int, Iterable[Int])] = rdd.groupBy(groupFunction2)
    rddGroup.collect().foreach(x=>println(x))

    val rdd1: RDD[String] = sc.makeRDD(List("1", "hello", "Spark", "Hello","Scala"),2)
    val rddGroup1: RDD[(Char, Iterable[String])] = rdd1.groupBy(x=>x.charAt(0))
    rddGroup1.collect().foreach(x=>println(x))



    //TODO 关闭环境
    sc.stop()


  }
}
